package com.shujia.flink.source

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo4MysqlSource {
  def main(args: Array[String]): Unit = {
    //创建flink环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //使用自定义的source读取mysql中的数据
    val mysqlDS: DataStream[String] = env.addSource(new MysqlSource())

    mysqlDS
      .map(stu => (stu.split("\t")(4), 1))
      .keyBy(_._1)
      .sum(1)
      .print()

    env.execute()

  }


  /**
   * 自定义source ,实现SourceFunction接口
   *
   */
  class MysqlSource extends SourceFunction[String] {
    /**
     * run： 用于读取外部数据的方法，只执行一次
     *
     * @param ctx ： 上下文对象，用于将读取到的数据发送到下游
     */
    override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
      /**
       * 使用jdbc读取mysql的数据，将读取到的数据发送到下游
       *
       */
      Class.forName("com.mysql.jdbc.Driver")

      //创建按链接
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata", "root", "123456")
      //编写查询数据的sql
      val stat: PreparedStatement = con.prepareStatement("select * from students")
      //执行查询
      val resultSet: ResultSet = stat.executeQuery()
      //解析数据
      while (resultSet.next()) {
        //通过字段名获取数据
        val id: Long = resultSet.getLong("id")
        val name: String = resultSet.getString("name")
        val age: Long = resultSet.getLong("age")
        val gender: String = resultSet.getString("gender")
        val clazz: String = resultSet.getString("clazz")

        //将每一条数据发送到下游
        ctx.collect(s"$id\t$name\t$age\t$gender\t$clazz")
      }

      //关闭链接
      stat.close()
      con.close()

    }

    //任务被取消的时候执行，一般用于回收资源
    override def cancel(): Unit = {}
  }
}
